/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package cn.doitedu;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;

/**
 * Skeleton for a Flink DataStream Job.
 *
 * <p>For a tutorial how to write a Flink application, check the
 * tutorials and examples on the <a href="https://flink.apache.org">Flink Website</a>.
 *
 * <p>To package your application into a JAR file for execution, run
 * 'mvn clean package' on the command line.
 *
 * <p>If you change the name of the main class (with the public static void main(String[] args))
 * method, change the respective entry in the POM.xml file (simply search for 'mainClass').
 */
public class DataStreamJob {

    public static void main(String[] args) throws Exception {
        // Sets up the execution environment, which is the main entry point
        // to building Flink applications.
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        KafkaSource<String> kafkaSource =
                KafkaSource.<String>builder()
                        .setTopics("doit47")
                        .setBootstrapServers("doitedu:9092")
                        // 优先选择上次提交的位移开始读；如果不存在之前提交的位移，则按照reset重置策略指定的位置读
                        .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
                        .setValueOnlyDeserializer(new SimpleStringSchema())
                        .setGroupId("g1")
                        .setClientIdPrefix("doit")
                        .build();

        env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "s")
                .map(line -> {
                    String[] split = line.split(",");
                    return new Bean(split[0], Integer.parseInt(split[1]));
                })
                .keyBy(bean -> bean.getTerm())
                .sum("score")
                .print();


        env.execute("doit47_niubility_job");
    }


    public static class Bean {
        private String term;
        private int score;

        public Bean(String term, int score) {
            this.term = term;
            this.score = score;
        }

        public Bean() {
        }

        public String getTerm() {
            return term;
        }

        public void setTerm(String term) {
            this.term = term;
        }

        public int getScore() {
            return score;
        }

        public void setScore(int score) {
            this.score = score;
        }

        @Override
        public String toString() {
            return "Bean{" +
                    "term='" + term + '\'' +
                    ", score=" + score +
                    '}';
        }
    }
}
